package per.chao.mqtt.domain.repo.subscribe;

import com.google.common.collect.Maps;
import com.google.common.collect.Sets;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.stereotype.Component;
import per.chao.mqtt.domain.facade.IMessageCache;
import per.chao.mqtt.domain.interfaces.repo.BaseSubscribeRepo;
import per.chao.mqtt.domain.shared.store.SubscribeStore;
import per.chao.mqtt.domain.util.TopicUtil;

import java.util.*;

/**
 * Description:
 *
 * @author W.Chao
 * @date 2021/6/20 16:31
 **/
@Component
public class SubscribeNotWildcardTopicRepo extends BaseSubscribeRepo<SubscribeStore> {
	private static final String SUBSCRIBE_NOT_WILDCARD_PREFIX = "subnotwildcard::";
	private static final String SUBSCRIBE_NOT_WILDCARD_CLIENT_PREFIX = "client::";

	@Autowired
	@Qualifier("redisCache")
	private IMessageCache iMessageCache;

	@Autowired
	@Qualifier("redisTopicCache")
	private IMessageCache iTopicCache;

	private String prefix(String prefix, String topic) {
		return TopicUtil.prefix(prefix, topic);
	}

	@Override
	public void put(String topic, String clientId, SubscribeStore subscribeStore) {
		// 将发布给clientId的消息存储在topic下
		String subNotWildcardTopicKey = prefix(SUBSCRIBE_NOT_WILDCARD_PREFIX, topic);
		if (iMessageCache.containsKey(subNotWildcardTopicKey)) {
			@SuppressWarnings("unchecked")
			Map<String, SubscribeStore> clientSubMsgMap = (Map<String, SubscribeStore>) iMessageCache.get(subNotWildcardTopicKey);
			clientSubMsgMap.put(clientId, subscribeStore);
			iMessageCache.put(subNotWildcardTopicKey, clientSubMsgMap);
		} else {
			Map<String, SubscribeStore> clientSubMsgMap = Maps.newHashMap();
			clientSubMsgMap.put(clientId, subscribeStore);
			iMessageCache.put(subNotWildcardTopicKey, clientSubMsgMap);
		}

		// 存储客户端和主题的订阅关系
		String clientSubTopicKey = prefix(SUBSCRIBE_NOT_WILDCARD_CLIENT_PREFIX, clientId);
		if (iTopicCache.containsKey(clientSubTopicKey)) {
			@SuppressWarnings("unchecked")
			Set<String> subTopicSets = (Set<String>) iTopicCache.get(prefix(SUBSCRIBE_NOT_WILDCARD_CLIENT_PREFIX, clientId));
			subTopicSets.add(topic);
			iTopicCache.put(clientSubTopicKey, subTopicSets);
		} else {
			Set<String> subTopicSets = Sets.newHashSet(topic);
			iTopicCache.put(clientSubTopicKey, subTopicSets);
		}
	}

	@SuppressWarnings("unchecked")
	@Override
	public SubscribeStore get(String topic, String clientId) {
		String subNotWildcardTopicKey = prefix(SUBSCRIBE_NOT_WILDCARD_PREFIX, topic);
		if (iMessageCache.containsKey(subNotWildcardTopicKey)) {
			return ((Map<String, SubscribeStore>) iMessageCache.get(subNotWildcardTopicKey)).get(clientId);
		}
		return null;
	}

	@Override
	@SuppressWarnings("unchecked")
	public void remove(String clientId) {
		Set<String> subTopics = (Set<String>) iTopicCache.get(clientId);
		if (subTopics == null || subTopics.size() == 0) {
			return;
		}
		for (String topic : subTopics) {
			String subNotWildcardTopicKey = prefix(SUBSCRIBE_NOT_WILDCARD_PREFIX, topic);
			if (iMessageCache.containsKey(subNotWildcardTopicKey)) {
				Map<String, SubscribeStore> clientSubMsgMap = (Map<String, SubscribeStore>) iMessageCache.get(subNotWildcardTopicKey);
				clientSubMsgMap.remove(clientId);
				iMessageCache.put(subNotWildcardTopicKey, clientSubMsgMap);
			}
		}

		String clientSubTopicKey = prefix(SUBSCRIBE_NOT_WILDCARD_CLIENT_PREFIX, clientId);
		if (iTopicCache.containsKey(clientSubTopicKey)) {
			iTopicCache.remove(clientId);
		}
	}

	@Override
	@SuppressWarnings("unchecked")
	public void remove(String topic, String clientId) {
		String subNotWildcardTopicKey = prefix(SUBSCRIBE_NOT_WILDCARD_PREFIX, topic);
		if (iMessageCache.containsKey(subNotWildcardTopicKey)) {
			Map<String, SubscribeStore> clientSubMsgMap = (Map<String, SubscribeStore>) iMessageCache.get(subNotWildcardTopicKey);
			clientSubMsgMap.remove(clientId);
			iMessageCache.put(subNotWildcardTopicKey, clientSubMsgMap);
		}

		String clientSubTopicKey = prefix(SUBSCRIBE_NOT_WILDCARD_CLIENT_PREFIX, clientId);
		if (iTopicCache.containsKey(clientSubTopicKey)) {
			iTopicCache.remove(clientId);
		}
	}

	@Override
	@SuppressWarnings("unchecked")
	public boolean containsKey(String topic, String clientId) {
		String subNotWildcardTopicKey = prefix(SUBSCRIBE_NOT_WILDCARD_PREFIX, topic);
		if (iMessageCache.containsKey(subNotWildcardTopicKey)) {
			Map<String, SubscribeStore> clientSubMsgMap = (Map<String, SubscribeStore>) iMessageCache.get(subNotWildcardTopicKey);
			return clientSubMsgMap.containsKey(clientId);
		}
		return false;
	}

	/**
	 * 获取所有主题下的所有订阅用户
	 *
	 * @return
	 */
	@SuppressWarnings("unchecked")
	public Map<String, Map<String, SubscribeStore>> all() {
		Set<String> topicKeys = iMessageCache.keys(SUBSCRIBE_NOT_WILDCARD_PREFIX);
		if (topicKeys != null && topicKeys.size() > 0) {
			Map<String, Map<String, SubscribeStore>> maps = Maps.newConcurrentMap();
			for (String topicKey : topicKeys) {
				Map<String, SubscribeStore> clientSubMsgMap = (Map<String, SubscribeStore>) iMessageCache.get(topicKey);
				if (clientSubMsgMap != null && clientSubMsgMap.size() > 0) {
					Map<String, SubscribeStore> newSubMsgMap = Maps.newHashMap(clientSubMsgMap);
					maps.put(topicKey.substring(SUBSCRIBE_NOT_WILDCARD_PREFIX.length()), newSubMsgMap);
				}
			}
			return maps;
		}
		return Maps.newConcurrentMap();
	}

	/**
	 * 获取该主题下所有订阅用户
	 *
	 * @param topic
	 * @return
	 */
	@SuppressWarnings("unchecked")
	public List<SubscribeStore> all(String topic) {
		String subNotWildcardTopicKey = prefix(SUBSCRIBE_NOT_WILDCARD_PREFIX, topic);
		if (iMessageCache.containsKey(subNotWildcardTopicKey)) {
			Map<String, SubscribeStore> clientSubMsgMap = (Map<String, SubscribeStore>) iMessageCache.get(subNotWildcardTopicKey);
			if (clientSubMsgMap != null) {
				return new ArrayList<>(clientSubMsgMap.values());
			}
		}
		return Collections.emptyList();
	}
}
